package com.hefei.garden.config.websocket.handler;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.hefei.garden.config.spring.SpringUtil;
import com.hefei.garden.config.websocket.pojo.MessageCarrier;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.AbstractWebSocketHandler;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author: devil
 * @Date: 2022/7/28 21
 * @Description:
 */
@Component
public class WebSocketHandler extends AbstractWebSocketHandler {

    private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketHandler.class);

    public static Map<String, WebSocketSession> sessionMap = new ConcurrentHashMap<>();

    public static Map<String, String> userMap = new ConcurrentHashMap<>();

    private ChannelTopic topic = SpringUtil.getBean(ChannelTopic.class);

    /**
     * 分布式 WebSocketSession
     * WebSocketSession 没有实现序列化接口，不能进行序列化，也就不能通过redis存储
     * 1. 使用redis的发布订阅
     * 2. 使用消息中间件
     * <p>
     * 本项目采用的是 redis 的发布订阅
     */
    private StringRedisTemplate redisTemplate = SpringUtil.getBean(StringRedisTemplate.class);

    /**
     * webSocket连接创建后调用
     */
    @Override
    public void afterConnectionEstablished(WebSocketSession session) {
        // 获取参数
        String user = String.valueOf(session.getAttributes().get("user"));
        String sessionId = session.getId();
        userMap.put(user, sessionId);
        sessionMap.put(sessionId, session);
        // 存储用户信息
        redisTemplate.execute((RedisCallback<Object>) connection -> {
            connection.select(0);
            connection.set(user.getBytes(), sessionId.getBytes());
            return null;
        });
        LOGGER.info("websocket 连接成功");
    }

    /**
     * 接收到消息会调用
     */
    @Override
    public void handleMessage(WebSocketSession session, WebSocketMessage<?> message) throws Exception {
        ObjectMapper jackson = new ObjectMapper();
        MessageCarrier messageCarrier;
        try {
            String messageInfo = message.getPayload().toString();
            messageCarrier = jackson.readValue(messageInfo, MessageCarrier.class);
        } catch (Exception e) {
            throw new RuntimeException("websocket 传输数据格式错误");
        }
        sendMessage(messageCarrier);
        LOGGER.info("处理消息成功");
    }

    /**
     * 连接出错会调用
     */
    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) {
        sessionMap.remove(session.getId());
    }

    /**
     * 连接关闭会调用
     */
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
        sessionMap.remove(session.getId());
    }

    @Override
    public boolean supportsPartialMessages() {
        return false;
    }

    /**
     * 后端发送消息，并实现分布式session
     * 群发暂未实现
     */
    public void sendMessage(MessageCarrier messageCarrier) throws IOException {
        String toUser = messageCarrier.getToUser();
        String message = messageCarrier.getContent();
        String sessionId = userMap.get(toUser);
        // 属于本服务器的用户
        if (StringUtils.isNotEmpty(sessionId)) {
            WebSocketSession session = sessionMap.get(sessionId);
            if (StringUtils.isNotEmpty(sessionId) && session != null && session.isOpen()) {
                session.sendMessage(new TextMessage(message));
                return;
            }
        }
        String toSessionId = redisTemplate.opsForValue().get(toUser);
        if (StringUtils.isBlank(toSessionId)) {
            return;
        }
        ObjectMapper jackson = new ObjectMapper();
        // 其他服务器的用户
        redisTemplate.convertAndSend(topic.getTopic(), jackson.writeValueAsString(messageCarrier));
    }

}

